package com.sui.bigdata.rtcadmin.async;

import com.alibaba.fastjson.JSONObject;
import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.feidee.fdcommon.model.StringResponse;
import com.feidee.fdcommon.util.HttpUtil;
import com.google.common.collect.Lists;
import com.sui.bigdata.rtcadmin.repository.model.JobConfig;
import com.sui.bigdata.rtcadmin.util.RedisUtils;
import com.sui.bigdata.rtcadmin.util.SendMsgUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

import static com.sui.bigdata.rtcadmin.constant.JobConstant.FLINK_WEB_PREFIX;
import static com.sui.bigdata.rtcadmin.constant.JobConstant.FLINK_WEB_SUFFIX;


/**
 * @author YongChen
 * @date 2019/12/13 18:56
 * @description
 * @email yong_chen@sui.com
 */
public class MemoryAlarm implements Runnable {

    protected static final Logger logger = LoggerFactory.getLogger(MemoryAlarm.class);

    private static String ID = "memory";

    private final Long ttl = 24 * 60 * 60L;

    private RedisUtils redisUtils;
    private JobConfig jobConfig;
    private String alarmUrl;
    private String memoryUrl;
    private String taskManager;
    private Integer todayMaxAlarmTimes;
    private Integer memoryAlarmThreshold;
    private Integer memoryRecoverThreshold;



    public MemoryAlarm(JobConfig jobConfig,
                       String alarmUrl,
                       String memoryUrl,
                       String taskManager,
                       RedisUtils redisUtils,
                       Integer todayMaxAlarmTimes,
                       Integer memoryAlarmThreshold,
                       Integer memoryRecoverThreshold
    ) {
        this.jobConfig = jobConfig;
        this.alarmUrl = alarmUrl;
        this.memoryUrl = memoryUrl;
        this.taskManager = taskManager;
        this.redisUtils = redisUtils;
        this.todayMaxAlarmTimes = todayMaxAlarmTimes;
        this.memoryAlarmThreshold = memoryAlarmThreshold;
        this.memoryRecoverThreshold = memoryRecoverThreshold;
    }

    @Override
    public void run() {
        try {
            if (StringUtils.isNotBlank(jobConfig.getAlarmGroupPhone())) {
                alarmUrl = alarmUrl + jobConfig.getAlarmGroupPhone();
            }
            JSONObject tmObject = JSONObject.parseObject(HttpUtil.get(memoryUrl).getResponseBody());
            Long maxMb = tmObject.getJSONObject("metrics").getLong("heapMax");
            Long usedMb = tmObject.getJSONObject("metrics").getLong("heapUsed");
            Double memoryUsedRate = Double.valueOf(usedMb) / maxMb * 100;
            String memoryPercent = Math.floor(Double.valueOf(usedMb) / maxMb * 100) + "%";

            SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
            String todayDate = formatter.format(new Date(System.currentTimeMillis()));

            String key = ID + jobConfig.getJobName() + taskManager + todayDate;
            Integer todaySendTimes = redisUtils.getInteger(key);

            if (memoryUsedRate > memoryAlarmThreshold && todaySendTimes % 2 == 0 && todaySendTimes < todayMaxAlarmTimes * 2) {
                SendMsgUtils.sendMemoryUsedMsg(alarmUrl, jobConfig.getJobName(), jobConfig.getDeveloper(), memoryPercent, taskManager, CustomConfiguration.getString(FLINK_WEB_PREFIX) + jobConfig.getAppId() + FLINK_WEB_SUFFIX);
                redisUtils.set(key, todaySendTimes + 1, ttl);
            }
            if (memoryUsedRate < memoryRecoverThreshold && todaySendTimes % 2 == 1) {
                SendMsgUtils.sendMemoryUsedMsg(alarmUrl, jobConfig.getJobName(), memoryPercent, taskManager);
                redisUtils.set(key, todaySendTimes + 1, ttl);
            }

            logger.info("the taskManager of {} called {} , memory used {}.", jobConfig.getJobName(), taskManager, memoryPercent);

        } catch (Exception e) {
            logger.warn("AlarmAlarm taskManager {} of {} 重试2次后执行失败: {}", taskManager, jobConfig.getJobName(), e);
        }
    }

    public static List<MemoryAlarm> build(JobConfig jobConfig,
                                          String alarmUrl,
                                          RedisUtils redisUtils,
                                          Integer todayMaxAlarmTimes,
                                          Integer memoryAlarmThreshold,
                                          Integer memoryRecoverThreshold) throws Exception {
        List<MemoryAlarm> backpressureAlarms = Lists.newArrayList();
        String webUrl = jobConfig.getWebUrl();
        String taskmanagersUrl = webUrl + "/taskmanagers";
        StringResponse response = null;
        for (int i = 0; i < 3; i++) {
            try {
                HttpGet httpGet = new HttpGet(taskmanagersUrl);
                RequestConfig requestConfig = RequestConfig.custom().setConnectTimeout(60000).build();
                httpGet.setConfig(requestConfig);
                response = HttpUtil.get(httpGet);
                break;
            } catch (Exception e) {
                if (i == 2) {
                    logger.warn("MemoryAlarm {} 重试2次后执行失败: {}", jobConfig.getJobName(), e);
                    return null;
                }
                Thread.sleep(200);
            }
        }
        if (null == response) {
            return null;
        }
        JSONObject tmsObject = JSONObject.parseObject(response.getResponseBody());
        for (Object taskmanager : tmsObject.getJSONArray("taskmanagers")) {
            JSONObject jsonNode = (JSONObject) taskmanager;
            String taskManager = jsonNode.get("id").toString();
            String memoryUrl = taskmanagersUrl + "/" + taskManager;
            backpressureAlarms.add(new MemoryAlarm(jobConfig,
                    alarmUrl,
                    memoryUrl,
                    taskManager,
                    redisUtils,
                    todayMaxAlarmTimes,
                    memoryAlarmThreshold,
                    memoryRecoverThreshold));
        }
        return backpressureAlarms;
    }
}
